Followable Apps

Note: Only stream and scheduled apps can be followed.

An app is called followable once it can be followed by other ones. Followable apps must produce data to trigger chain reaction of following app runs. See Produce messages and Insert data sections for instructions on how to produce messages.

from corva import Api, Cache, ScheduledDataTimeEvent, StreamTimeEvent, scheduled, stream

@scheduled #1
def followable_scheduled_app(event: ScheduledDataTimeEvent, api: Api, cache: Cache):
    data = [
        {
            'asset_id': event.asset_id,
            'version': 1,
            'timestamp': 0,
            'data': {'answer': 10},
        },
        {
            'asset_id': event.asset_id,
            'version': 1,
            'timestamp': 60,
            'data': {'answer': 11},
        },
    ] #2

    #3
    api.insert_data(
        provider='my-provider',
        dataset='quiz-answers',
        data=data,
    ) #4
    api.produce_messages(data=data) #5

    #6
    api.insert_data(
        provider='my-provider',
        dataset='quiz-answers',
        data=data,
        produce=True, #7
    )


@scheduled #8
def following_scheduled_app(event: ScheduledDataTimeEvent, api: Api, cache: Cache):
    data = api.get_dataset(
        provider='my-provider',
        dataset='quiz-answers',
        query={'asset_id': event.asset_id, 'company_id': event.company_id},
        sort={'timestamp': 1},
        limit=2,
    ) #9

    assert [datum['data'] for datum in data] == [
        {'answer': 10},
        {'answer': 11},
    ]


@stream #10
def following_stream_app(event: StreamTimeEvent, api: Api, cache: Cache):
    assert [record.data for record in event.records] == [
        {'answer': 10},
        {'answer': 11},
    ] #11

1. The followable app.

2. Data documents that should be saved and produced.

3. Slow method which uses two api calls.

4. First save the data to the dataset. See Insert data for details.

5. Then produce the messages. See Produce messages for details.

6. Recommended method which uses single api call.

7. Enable this flag to save and produce messages at once. See Insert data for details.

8. Example scheduled following app. It will be invoked as soon as there is some time interval worth of data.

9. Scheduled apps should query the data as unlike stream ones they don’t receive it in the event.

10. Example stream following app. It will be invoked as soon as new messages are produced.

11. Stream apps receive data straight from the event. No api calls needed.